package tsif.tcluster.netty;

import java.util.concurrent.TimeUnit;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import tsif.tcluster.netty.TClusterNettySession.TClusterNettySessionHandler;
import tsif.tcluster.netty.codec.TClusterNettyDecoder;
import tsif.tcluster.netty.codec.TClusterNettyEncoder;
import tsif.tcluster.netty.codec.TClusterNettyMessage;
import tsif.tcluster.netty.impl.NettyClientMgr;
import tsif.tcluster.rpc.TClusterRpcInfoMsg;
import tsif.tcluster.rpc.TClusterRpcProtocol;
import tsif.tcluster.rpc.TClusterRpcUtils;

/**
 * 集群连接管理器<br>
 * 用netty管理集群环境下的连接<br>
 */
public class TClusterNettyClient extends NettyClientMgr {
	protected static final AttributeKey<ConnectSession> TSESSION_KEY = AttributeKey.valueOf("TSESSION");
	protected static final AttributeKey<TClusterClientHandler> THEADER_KEY = AttributeKey.valueOf("THEADER");
	protected static final TClusterNettyMessage heartbeatMsg;

	static {
		// 心跳消息
		heartbeatMsg = new TClusterNettyMessage();
		heartbeatMsg.setCode(TClusterRpcProtocol.RPC_KEEP);
	}

	protected TClusterNettyContext context;

	public TClusterNettyClient(TClusterNettyContext context) {
		this.context = context;
	}

	@Override
	protected boolean initOption(Bootstrap bootstrap) {
		final int BUFF_SIZE = TClusterNettyServer.BUFF_SIZE;

		bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
		bootstrap.option(ChannelOption.TCP_NODELAY, true);
		bootstrap.option(ChannelOption.SO_RCVBUF, BUFF_SIZE);
		bootstrap.option(ChannelOption.SO_SNDBUF, BUFF_SIZE);
		bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10 * 1000);
		bootstrap.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, (int) (BUFF_SIZE * 1));
		bootstrap.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, (int) (BUFF_SIZE * 0.5));
		return true;
	}

	@Override
	protected ChannelHandler createHandler() {
		return new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				// create handler
				TClusterClientHandler handler = new TClusterClientHandler();
				ch.attr(THEADER_KEY).set(handler);

				// init channel
				ChannelPipeline pipeline = ch.pipeline();
				pipeline.addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS));
				pipeline.addLast("decoder", new TClusterNettyDecoder());
				pipeline.addLast("encoder", new TClusterNettyEncoder());
				pipeline.addLast("handler", handler);
			}
		};
	}

	/** 连接服务器 **/
	public ConnectSession connectSession(int contextId, String host, int port) {
		ConnectSession session = new ConnectSession(this, contextId, host, port);
		if (!session.connect()) {
			throw new RuntimeException("connect server error! " + host + " " + port);
		}
		return session;
	}

	@Override
	protected NioEventLoopGroup createChildGroup(String name, int threads) {
		NioEventLoopGroup eventLoopGroup = context.getEventLoopGroup();
		if (eventLoopGroup != null) {
			return eventLoopGroup;
		}
		return super.createChildGroup(name, threads);
	}

	/** 连接session **/
	public static class ConnectSession extends TClusterNettySession {
		protected TClusterNettyClient client;
		protected String host;
		protected int port;

		public ConnectSession(TClusterNettyClient client, int targetContextId, String host, int port) {
			super(client.context, targetContextId, null);
			this.client = client;
			this.host = host;
			this.port = port;
		}

		/** 执行连接, 连接完毕发送消息验证 **/
		public boolean connect() {
			try {
				// connect
				this.ch = client.connect(host, port);
				if (ch == null) {
					return false;
				}
				// bind
				TClusterClientHandler handler = ch.attr(THEADER_KEY).get();
				handler.setSession(this);

				// write info
				TClusterRpcInfoMsg infomsg = new TClusterRpcInfoMsg();
				infomsg.setSourceContextId(this.context.getContextId());
				infomsg.setTargetContextId(this.targetContextId);

				// write connect
				TClusterNettyMessage packet = new TClusterNettyMessage();
				packet.setCode(TClusterRpcProtocol.RPC_INFO);
				packet.setData(TClusterRpcUtils.toByte(infomsg, infomsg.getClass()));
				this.ch.writeAndFlush(packet);
				return true;
			} catch (Exception e) {
				// return false; // connect error
				throw new RuntimeException(e);
			}
		}

		@Override
		public String toString() {
			return "ConnectSession [host=" + host + ", port=" + port + "]";
		}
	}

	class TClusterClientHandler extends TClusterNettySessionHandler {
		@Override
		public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
			if (evt instanceof IdleStateEvent) {
				IdleStateEvent event = (IdleStateEvent) evt;
				if (event.state() == IdleState.WRITER_IDLE) {
					// 闲置发送心跳
					ctx.channel().writeAndFlush(heartbeatMsg);
					// System.out.println("闲置发送心跳!" + (new Date()) + ctx.channel());
				}
			} else {
				super.userEventTriggered(ctx, evt);
			}
		}
	}

}